package study.cdc.debezium.support.debezium;

import io.debezium.engine.ChangeEvent;
import io.debezium.engine.DebeziumEngine;
import io.debezium.engine.format.Json;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import study.cdc.debezium.support.connector.ChangeEventConsumer;
//import study.cdc.debezium.support.connector.JsonHandler;

/**
 * Debezium Engine配置: https://debezium.io/documentation/reference/1.5/development/engine.html#engine-properties
 * <p>
 * https://mp.weixin.qq.com/s/_67XXbPAawegCP08W8FHIQ
 * https://blog.csdn.net/alionsss/article/details/116501918
 * https://www.modb.pro/db/429533
 */
@Slf4j
//@Configuration
public class DebeziumServerConfig {

    /**
     * 应用程序需要为运行的 Mysql Connector 启动一个 Debezium 引擎，这个引擎会以异步线程的形式运行，
     * 它包装了整个 Mysql Connector 连接器的生命周期
     */
    @Bean
    public DebeziumServer debeziumServer(io.debezium.config.Configuration configuration) {
        log.info(">>>>>>>>>> init DebeziumRunner");

        //step-1: Debezium Engine
        DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
                //配置
                .using(configuration.asProperties())
                //Completion Callback
                .using((success, message, error) -> {
                    log.info(">>>>>>>>>>>>> cal");
                    // 强烈建议加上此部分的回调代码，方便查看错误信息
                    if (!success && error != null) {
                        // 报错回调
                        System.out.println("----------error------");
                        System.out.println(message);
                    }
                })
                //消费数据
                .notifying(new ChangeEventConsumer())
                .build();

        //step-2: Debezium Server
        DebeziumServer server = new DebeziumServer();
        server.setEngine(engine);

        return server;
    }
}
